package demo.environment;

import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.DeploymentOptions;
import org.apache.flink.core.execution.*;
import org.apache.flink.streaming.api.graph.StreamGraph;
import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.FlinkException;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

import static org.apache.flink.util.Preconditions.checkNotNull;

public class StreamExecutionEnvironmentDemo {

    private final Configuration configuration;
    private final PipelineExecutorServiceLoader executorServiceLoader = new DefaultExecutorServiceLoader();
    private final ClassLoader userClassloader = null;

    public StreamExecutionEnvironmentDemo(Configuration configuration) {
        this.configuration = configuration;
        /*this.executorServiceLoader = executorServiceLoader;
        this.userClassloader = userClassloader;*/
    }


    /**
     * 触发程序执行。该环境将执行导致“sink”操作的程序的所有部分。例如，接收操作是打印结果或将结果转发到消息队列。
     * @param streamGraph—表示转换的流图
     * @return 作业执行的结果，包含运行时间和累加器。
     * @throws Exception—在作业执行过程中发生。
     */
    @Internal
    public JobExecutionResult execute(StreamGraph streamGraph) throws Exception {
        final JobClient jobClient = executeAsync(streamGraph);

        try {
            final JobExecutionResult jobExecutionResult;

            if (configuration.getBoolean(DeploymentOptions.ATTACHED)) {
                jobExecutionResult = jobClient.getJobExecutionResult(userClassloader).get();
            } else {
                jobExecutionResult = new DetachedJobExecutionResult(jobClient.getJobID());
            }

            return jobExecutionResult;
        } catch (Throwable t) {
            ExceptionUtils.rethrowException(t);
            return null;
        }
    }

    /**
     * 异步触发程序执行。该环境将执行导致“sink”操作的程序的所有部分。例如，接收操作是打印结果或将结果转发到消息队列。
     * @param streamGraph—表示转换的流图
     * @return 可以用来与提交的作业通信的JobClient，在提交成功时完成。
     * @throws Exception—在作业执行过程中发生。
     */
    @Internal
    public JobClient executeAsync(StreamGraph streamGraph) throws Exception {
        checkNotNull(streamGraph, "StreamGraph cannot be null.");
        checkNotNull(configuration.get(DeploymentOptions.TARGET), "No execution.target specified in your configuration file.");

        final PipelineExecutorFactory executorFactory =
                executorServiceLoader.getExecutorFactory(configuration);

        checkNotNull(
                executorFactory,
                "Cannot find compatible factory for specified execution.target (=%s)",
                configuration.get(DeploymentOptions.TARGET));

        CompletableFuture<JobClient> jobClientFuture = executorFactory
                .getExecutor(configuration)
                .execute(streamGraph, configuration);

        try {
            JobClient jobClient = jobClientFuture.get();
            return jobClient;
        } catch (ExecutionException executionException) {
            final Throwable strippedException = ExceptionUtils.stripExecutionException(executionException);
            throw new FlinkException(
                    String.format("Failed to execute job '%s'.", streamGraph.getJobName()),
                    strippedException);
        }
    }
}
